﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;

namespace Ec.NETCoreFramework.Common.MQ
{
    /// <summary>
    /// rabbitmq
    /// </summary>
    public class RabbitMQInvoker
    {
        #region Identity
        private string _hostName = string.Empty;
        private string _userName = string.Empty;
        private string _pwd = string.Empty;
        private string _virtualHost = string.Empty;
        /// <summary>
        /// 
        /// </summary>
        /// <param name="hostName"></param>
        /// <param name="userName"></param>
        /// <param name="pwd"></param>
        /// <param name="virtualHost"></param>
        public RabbitMQInvoker(string hostName, string userName, string pwd, string virtualHost)
        {
            _hostName = hostName;
            _userName = userName;
            _pwd = pwd;
            _virtualHost = virtualHost;
        }
        #endregion

        #region Init
        private static object RabbitMQInvoker_InitLock = new object();
        private static IConnection _CurrentConnection = null;//链接做成单例重用--channel是新的
        /// <summary>
        /// 初始化连接
        /// </summary>
        private void InitConnection()
        {
            if (_CurrentConnection == null || !_CurrentConnection.IsOpen)
            {
                lock (RabbitMQInvoker_InitLock)
                {
                    if (_CurrentConnection == null || !_CurrentConnection.IsOpen)
                    {
                        var factory = new ConnectionFactory()
                        {
                            HostName = _hostName,
                            Password = _pwd,
                            UserName = _userName,
                            VirtualHost = _virtualHost
                        };
                        _CurrentConnection = factory.CreateConnection();
                    }
                }
            }
        }
        #endregion

        /// <summary>
        /// 只管exchange---
        /// 4种路由类型？
        /// 
        /// Send前完成交换机初始化
        /// </summary>
        /// <param name="exchangeName"></param>
        /// <param name="body">byte</param>
        public void Send(string exchangeName, byte[] body)
        {
            if (_CurrentConnection == null || !_CurrentConnection.IsOpen)
            {
                this.InitConnection();
            }
            using (var channel = _CurrentConnection.CreateModel())//开辟新的信道通信
            {
                try
                {
                    channel.ExchangeDeclare(exchange: exchangeName, type: ExchangeType.Fanout, durable: true, autoDelete: false, arguments: null);

                    channel.TxSelect();//开启Tx事务---RabbitMQ协议级的事务-----强事务

                    channel.BasicPublish(exchange: exchangeName,
                                         routingKey: string.Empty,
                                         basicProperties: null,
                                         body: body);
                    channel.TxCommit();//提交
                    Console.WriteLine($" [x] Sent {body.Length}");
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                    channel.TxRollback(); //事务回滚--前面的所有操作就全部作废了。。。。
                }
            }
        }

        /// <summary>
        /// 固定无消费队列名字---转移到目标队列---定好时间
        /// </summary>
        /// <param name="targetExchangeName"></param>
        /// <param name="body"></param>
        /// <param name="delaySecond"></param>
        public void SendDelay(string targetExchangeName, byte[] body, int delaySecond)
        {
            if (_CurrentConnection == null || !_CurrentConnection.IsOpen)
            {
                this.InitConnection();
            }
            using (var channel = _CurrentConnection.CreateModel())//开辟新的信道通信
            {
                try
                {
                    var delayName = "RWXKJ_DelayQueue";
                    var delayExchangeName = "RWXKJ_DelayExchange";

                    //普通交换器
                    channel.ExchangeDeclare(delayExchangeName, "fanout", true, false, null);
                    //参数设置
                    var args = new Dictionary<string, object>();
                    args.Add("x-message-ttl", delaySecond * 1000);//TTL 毫秒
                    args.Add("x-dead-letter-exchange", targetExchangeName);//DLX
                    args.Add("x-dead-letter-routing-key", "routingkey");//routingKey
                    channel.QueueDeclare(delayName, true, false, false, args);
                    channel.QueueBind(queue: delayName,exchange: delayExchangeName,routingKey: string.Empty,arguments: null);


                    channel.TxSelect();//开启Tx事务---RabbitMQ协议级的事务-----强事务
                    var properties = channel.CreateBasicProperties();

                    channel.BasicPublish(exchange: delayExchangeName,
                                         routingKey: string.Empty,
                                         basicProperties: properties,
                                         body: body);
                    channel.TxCommit();//提交
                    Console.WriteLine($" [x] Sent {body.Length}");
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                    channel.TxRollback(); //事务回滚--前面的所有操作就全部作废了。。。。
                }
            }
        }

        #region Receive
        /// <summary>
        /// 注册处理动作
        /// </summary>
        /// <param name="consumeInterface"></param>
        /// <param name="func"></param>
        public  void RegistReciveAction(TaskInterface consumeInterface)
        {
            if (_CurrentConnection == null || !_CurrentConnection.IsOpen)
            {
                this.InitConnection();
            }

            var queueName = consumeInterface.QueueName();
            var exchangeName = consumeInterface.ExchangeName;

            using (var channel = _CurrentConnection.CreateModel())
            {
                channel.ExchangeDeclare(exchange: exchangeName, 
                    type: ExchangeType.Fanout,
                    durable: true, 
                    autoDelete: false);
                channel.QueueDeclare(queue: queueName, 
                    durable: true, 
                    exclusive: false, 
                    autoDelete: false);
                channel.QueueBind(queue: queueName,
                    exchange: exchangeName, 
                    routingKey: string.Empty);

                var consumer = new EventingBasicConsumer(channel);
                channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true);
                consumer.Received += (sender, ea) =>
                {
                    var body = ea.Body.ToArray();
                    var result = consumeInterface.ConsumeEvent(body);
                    if (result.Result)
                    {
                        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);//确认已消费
                    }
                    else if (consumeInterface.IsAck)
                    {
                        channel.BasicReject(deliveryTag: ea.DeliveryTag, requeue: true);//放回队列--重新包装信息，放入其他队列
                    }
                };
                channel.BasicConsume(queue: consumeInterface.QueueName(),
                                     autoAck: false,//不ACK
                                     consumer: consumer);
                Console.WriteLine($" Register Consumer To {consumeInterface.ExchangeName}-{consumeInterface.QueueName()}");
                Console.ReadLine();
                Console.WriteLine($" After Register Consumer To {consumeInterface.ExchangeName}-{consumeInterface.QueueName()}");
            }
        }
        #endregion
    }
}
